package com.atguigu.worktest;

import com.atguigu.bean.OrderEvent;
import com.atguigu.bean.TxEvent;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashSet;
import java.util.Set;

public class Flink04_Order_Connect {
    public static void main(String[] args) {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",10000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(2);

        //需要解决的问题，就是可能订单信息已经提交完成，但是支付信息明明已经支付了却没有收到确认
        //读取订单流
        DataStreamSource<String> orderfDatas = env.readTextFile("input/OrderLog.csv");

        //读取交易流
        DataStreamSource<String> receiptDatas = env.readTextFile("input/ReceiptLog.csv");

        //对数据进行处理，因为我们要对两个不同来源的数据进行处理
        //所以需要对双流合一
        ConnectedStreams<String, String> connectDatas = orderfDatas.connect(receiptDatas);

        //对数据进行处理,这里必须要进行keyBy操作，不然两个数据分别处理，那么就可能id值一样的数据交会不到一起去，除非并行度设置为1
        //那么就需要提前对数据进行一个处理之后再进行双流合并
        connectDatas.process(new CoProcessFunction<String, String, String>() {
            //定义数据结构存储订单和交易进来的数据
            Set<String> orderSet  = new HashSet<>();
            Set<String> txSet  = new HashSet<>();


            @Override
            public void processElement1(String value, Context ctx, Collector<String> out) throws Exception {
                //这里处理传进来的订单流数据
                String[] datas = value.split(",");
                //对数据进行封装
                OrderEvent singleOrder = new OrderEvent(Long.valueOf(datas[0]), datas[1], datas[2], Long.valueOf(datas[3]));

                //这里得到了一条订单数据，首先需要进行的是对临时交易集合中的数据进行查询
                if (txSet.contains(singleOrder.getTxId())) {
                    //这里查询到了，说明order数据晚来了，那么对数据进行处理
                    out.collect("txId:"+singleOrder.getTxId()+",我是订单数据,晚来了，但是和交易数据对上了！");
                    //数据对上了，那么之前在txSet中的数据就没有用了，删除
                    txSet.remove(singleOrder.getTxId());
                }else{
                    //这里处理的逻辑是我来早了，自己把自己放在临时orderSet中
                    orderSet.add(singleOrder.getTxId());
                }

            }

            @Override
            public void processElement2(String value, Context ctx, Collector<String> out) throws Exception {
                //这里处理传进来的交易流数据
                String[] datas = value.split(",");
                //对数据进行封装
                TxEvent singleTxData = new TxEvent(datas[0], datas[1], Long.valueOf(datas[2]));



                //这里得到了一条订单数据，首先需要进行的是对临时交易集合中的数据进行查询
                if (orderSet.contains(singleTxData.getTxId())) {
                    //这里查询到了，说明tx数据晚来了，那么对数据进行处理
                    out.collect("txId:"+singleTxData.getTxId()+",我是交易数据,晚来了，但是和订单数据对上了！");
                    //数据对上了，那么之前在orderSet中的数据就没有用了，删除
                    orderSet.remove(singleTxData.getTxId());
                }else{
                    //这里处理的逻辑是我来早了，自己把自己放在临时orderSet中
                    txSet.add(singleTxData.getTxId());
                }

            }
        }).print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
